package com.diver.flinkdemo;

import com.diver.flinkdemo.controller.DalDataController;
import com.diver.flinkdemo.dataSoucer.MyDataSource;
import com.diver.flinkdemo.entity.DalData;
import com.diver.flinkdemo.entity.DevUploadData;
import com.diver.flinkdemo.entity.xxt.IssuedPermissionRecord;
import com.diver.flinkdemo.entity.xxt.LastLocation;
import com.diver.flinkdemo.entity.xxt.LocationInfo;
import com.diver.flinkdemo.service.LocationInfoHandleService;
import com.diver.flinkdemo.sink.DevUploadDataSink;
import com.diver.flinkdemo.sink.IssuedPermissionRecordSink;
import com.diver.flinkdemo.sink.LastLocationSink;
import com.diver.flinkdemo.sink.LocationInfoSink;
import com.diver.flinkdemo.utils.DateUtil;
import com.diver.flinkdemo.utils.DynamicDataSourceUtil;
import com.diver.flinkdemo.utils.JedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;

import java.util.Date;
import java.util.Properties;

/**
 * @author LuJW
 * @date 2023/3/17 16:30
 */
@Slf4j
public class LocationInfoSaveJob {

//    private static DalDataController dalDataController;
    private static String bootstrapServers = "172.16.100.67:9092";
    private static String topic_upload_data = "aiot-msg";

    public static void main(String[] args){
        try {
            DynamicDataSourceUtil sourceUtil = new DynamicDataSourceUtil();
            sourceUtil.init();

            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//             exactly-once 语义保证整个应用内端到端的数据一致性
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//             开启检查点并指定检查点时间间隔为5s
            env.enableCheckpointing(60000); // checkpoint every 5000 msecs

            KafkaSource<String> source =
                    KafkaSource.<String>builder().setBootstrapServers("172.16.100.67:9092").setTopics("aiot-msg")
                            .setGroupId("flinkdev").setStartingOffsets(OffsetsInitializer.latest())//OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST)
                            .setValueOnlyDeserializer(new SimpleStringSchema()).build();

            DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source") // 指定从最新offset开始消费;
                    .setParallelism(1);
            //3.Locationinfo,Lastlocation数据处理之后入库
            //（1）数据传入dalSendApp进行处理后再过滤出高德解析后的数据（PARSED_LOCATION）
            SingleOutputStreamOperator<LocationInfo> map = stream.filter(new FilterFunction<String>() {
                @Override
                public boolean filter(String s) throws Exception {
                    log.info("过滤出非TASK_TRACKING类型的数据");
                    DalData dalData = DalData.convert(s);
                    return "PARSED_LOCATION".equals(dalData.getType()) || "XSZ_UP_MSG".equals(dalData.getType());
                }
            }).map(new MapFunction<String, LocationInfo>() {
                @Override
                public LocationInfo map(String value) throws Exception {
                    log.info("对非TASK_TRACKING类型的数据进行处理");
                    DalData dalData = DalData.convert(value);
                    LocationInfoHandleService service = new LocationInfoHandleService();
                    LocationInfo locationInfo = service.getLocation(new LocationInfo(), dalData);
                    return locationInfo;
                }
            });
            //（2） locationInfo lastLocation 数据入库
            map.addSink(new LocationInfoSink());

            env.execute("LocationInfoSaveJob");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

//    public
}
